package sc.analysis.vortex;



import java.io.IOException;
import java.util.LinkedList;
import java.util.List;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import sc.util.FinancialNumbers;
import sc.writable.PriceData;
import sc.writable.WritableArrayListFloat;



/**
 * Mapper of the Vortex analysis. Calculates Vortex Up, Vortex Down and the
 * number of days since the indicators have crossed.
 * 
 * @author Quentin Rossé
 *
 */
public class VortexMapper extends Mapper<Text, PriceData, Text, VortexData>{
	
	/**
	 * Periods (in days) for the indicator
	 */
	private static final int[] PERIODS = {21, 55, 120, 200};
	
	/**
	 * maximum number of reading for a period
	 */
	private static final int MAX_READINGS = 30;
	
	/**
	 * data that will be written to the reducer
	 */
	private static VortexData vd = new VortexData();
	
	
	public void map(Text key, PriceData value, Context context) throws IOException, InterruptedException{
		
		WritableArrayListFloat dayLows = value.getDayLowPrices();
		WritableArrayListFloat dayHighs = value.getDayHighPrices();
		WritableArrayListFloat close = value.getClosePrices();

		LinkedList<Float> TrueRange = new LinkedList<Float>();
		LinkedList<Float> VMUp = new LinkedList<Float>();
		LinkedList<Float> VMDown = new LinkedList<Float>();
		
		float VIUp[][] = new float[PERIODS.length][MAX_READINGS];
		float VIDown[][] = new float[PERIODS.length][MAX_READINGS];
		
		//calculation of the VMUp and VMDown (they are the same for all 
		//perdiods)
		for(int i=close.size()-1; i>0; i--){
			TrueRange.add(FinancialNumbers.trueRange(dayHighs.getFloat(i), 
                                                dayLows.getFloat(i),
                                                close.getFloat(i-1)));
			VMUp.add(Math.abs(dayHighs.getFloat(i) - dayLows.getFloat(i-1)));
			VMDown.add(Math.abs(dayLows.getFloat(i) - dayHighs.getFloat(i-1)));
		}
		
		//calculation of VIUp and VIDown for all the periods
		for(int i=0; i<PERIODS.length; i++){
			for(int j=0; j<MAX_READINGS; j++){
				if(j+PERIODS[i] <= TrueRange.size()){
					VIUp[i][j] = vi(TrueRange, VMUp, PERIODS[i], j);
					VIDown[i][j] = vi(TrueRange, VMDown, PERIODS[i], j);
				}
				else{
					VIUp[i][j] = Float.NaN;
					VIDown[i][j] = Float.NaN;
					break;
				}
			}
		}
		
		//calculation of the useful informations for the next phase and writing
		//the results
		//true if VIUp is above VIDown
		boolean viUpUp;
		for(int i=0; i<PERIODS.length; i++){
			//if there's not enough data for this indicator, we don't go to the
			//next
			if(Float.isNaN(VIUp[i][0])){
				continue;
			}
			
			viUpUp = VIUp[i][0] >= VIDown[i][0];
			for(int j=1; j<=MAX_READINGS; j++){
				if(j == MAX_READINGS || Float.isNaN(VIUp[i][j])){
					vd.setAll(PERIODS[i], VIDown[i][0], VIUp[i][0], -1);
					context.write(key, vd);
					break;
				}
				if(viUpUp){
					if(VIDown[i][j] > VIUp[i][j]){
						vd.setAll(PERIODS[i], VIDown[i][0], VIUp[i][0], j-1);
						context.write(key, vd); 
						break;
					}
				}
				else{
					if(VIUp[i][j] > VIDown[i][j]){
						vd.setAll(PERIODS[i], VIDown[i][0], VIUp[i][0], j-1);
						context.write(key, vd); 
						break;
					}
				}
			}
		}
		
	}
	
	/**
	 * Calculates the VI (up or down) with the VM
	 * 
	 * @param trueRange list of True Range values
 	 * @param vm list of VM values
	 * @param period number of days for the indicator
	 * @param ith 0 --> last value, 1 --> value of the day before, ...
	 * 
	 * @return VI (up if vm if VM up or down if vm if VM down)
	 */
	private float vi(List<Float> trueRange,
			         List<Float> vm,
			         int period,
			         int ith){
		float sumTR = 0f;
		float sumVM = 0f;
		
		for(int i=ith; i<period; i++){
			sumTR += trueRange.get(i);
			sumVM += vm.get(i);
		}
		return sumVM/sumTR;
	}
	
	@Override
	public void setup(Context context){}
	
	@Override
	public void cleanup(Context context){}
	
}
